package io.xxx.xbutler.core.event.job;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import io.xxx.xbutler.constant.Topics;
import io.xxx.xbutler.data.TaskReceiverMapper;
import io.xxx.xbutler.domain.task.TaskReceiver;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.MessageId;
import org.quartz.JobExecutionContext;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;

/**
 * @see io.xxx.xbutler.core.event.listen.MessagePushListener
 */
@Slf4j
@Component
public class MessagePushJob extends QuartzJobBean {

    @Resource
    private PulsarTemplate<Long> pulsarTemplate;

    @Resource
    private TaskReceiverMapper taskReceiverMapper;

    @Override
    protected void executeInternal(@Nonnull JobExecutionContext context) {
        LambdaQueryWrapper<TaskReceiver> wrapper = new LambdaQueryWrapper<TaskReceiver>()
                .le(TaskReceiver::getPushTime, LocalDateTime.now())
                .eq(TaskReceiver::getPushed, false);
        List<TaskReceiver> taskReceivers = taskReceiverMapper.selectList(wrapper);
        for (TaskReceiver taskReceiver : taskReceivers) {
            taskReceiver.setPushed(true)
                    .setUpdatedTime(LocalDateTime.now());
            MessageId messageId = pulsarTemplate.send(Topics.MESSAGE_PUSH, taskReceiver.getId());
            taskReceiverMapper.updateById(taskReceiver);
            log.info("任务消息已发送[taskId:{}, receiverType:{}, receiverId:{}, taskReceiverId:{}, messageId:{}]",
                    taskReceiver.getTaskId(), taskReceiver.getReceiverType(), taskReceiver.getReceiverId(), taskReceiver.getId(), messageId);
        }
    }
}
